package com.lagou.kafka.demo.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description: 消费者拦截器
 * @date 2022/6/13 8:27
 */
public class OneInterceptor implements ConsumerInterceptor<Integer,String> {


    @Override
    public ConsumerRecords<Integer, String> onConsume(ConsumerRecords<Integer, String> records) {
        // poll返回结果之前最后要调用的方法
        System.out.println("One -- 开始");
        // 消息不做处理,直接返回
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 消费者提交偏移量的时候,经过该方法
        System.out.println("One -- 结束");
    }

    @Override
    public void close() {
        // 用于关闭拦截器用到的资源,如打开文件，连接数据库
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 用于获取消费者的设置参数
        configs.forEach((k,v)->{
            System.out.println(k+"\t"+v);
        });

    }
}
